/*
 * Package il.ac.biu.cs.grossmm.impl.activeData
 * File NotificationPointImpl.java
 * 
 * This is a part of presence service framework prototype. 
 * See javadoc for more information.
 *  
 * Copyright (C) 2006 Grossmann Mark
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public License
 * as published by the Free Software Foundation; either version 2
 * of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.

 * You should have received a copy of the GNU Lesser General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 */
package il.ac.biu.cs.grossmm.impl.activeData;

import il.ac.biu.cs.grossmm.api.data.Node;
import il.ac.biu.cs.grossmm.api.data.NodeType;
import il.ac.biu.cs.grossmm.api.data.UpdateType;
import il.ac.biu.cs.grossmm.api.flow.NoSuchSubscription;
import il.ac.biu.cs.grossmm.api.flow.NodeEventInspector;
import il.ac.biu.cs.grossmm.api.flow.NodeManipulator;
import il.ac.biu.cs.grossmm.api.flow.NotificationPoint;
import il.ac.biu.cs.grossmm.api.flow.NotificationProvider;
import il.ac.biu.cs.grossmm.api.flow.Notifier;
import il.ac.biu.cs.grossmm.api.flow.Status;
import il.ac.biu.cs.grossmm.api.flow.Subscriber;
import il.ac.biu.cs.grossmm.api.flow.Track;
import il.ac.biu.cs.grossmm.api.keys.Key;
import il.ac.biu.cs.grossmm.api.keys.KeyPattern;
import il.ac.biu.cs.grossmm.impl.keys.PatternMatcher;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;

public class NotificationPointImpl<N> implements NotificationPoint<N>,
		NotificationProvider<N> {

	private static Logger logger = Logger.getLogger("NotificationPointImpl");

	// static {
	// logger.setLevel(Level.FINE);
	// }

	final DataLayer<N> dataLayer;

	final KeyPattern pattern;

	final PatternMatcher matcher;

	final Notifier notifier;

	final NodeType nodeType;

	final double priority;

	static final NodeEventInspector eventInspector = new MemEventInspector();

	private Map<Key, SubscriptionImpl<N>> subscriptions;

	Key pendingKey;

	Subscriber<N> pendingSubscriber;

	Lock pendingLock;

	Lock subLock;

	private SubscriptionImpl<N> pendingSubscription;

	public NotificationPointImpl(DataLayer<N> dataLayer, Notifier notifier,
			KeyPattern pattern, NodeType nodeType, double priority) {
		this.dataLayer = dataLayer;
		this.notifier = notifier;
		this.pattern = pattern;
		this.matcher = new PatternMatcher(pattern);
		this.nodeType = nodeType;
		this.priority = priority;
		this.subscriptions = new ConcurrentHashMap<Key, SubscriptionImpl<N>>();

		this.pendingLock = new ReentrantLock();
		this.subLock = new ReentrantLock();
	}

	public NodeManipulator writeLock(Node node) {
		if (logger.isLoggable(Level.FINE)) {
			logger.fine(MemInspector.INSTANCE.toString(node));
		}

		// TODO lock
		return dataLayer.adm.memManipulator;
	}

	public void writeUnlock(Node node) {
		MemRootNode<N> n = (MemRootNode<N>) node.getRoot();

		if (n.getUpdateType(1) == UpdateType.UNCHANGED) {
			if (logger.isLoggable(Level.FINE))
				logger.fine("unchanged: "
						+ MemInspector.INSTANCE.toString(node));
			return; // no change
		}

		if (n.getKey() == pendingKey) {
			// works correcty, however in some cases may
			// lock on a wrong pending key which does
			// not result in an incorrect behaviour
			pendingLock.lock();
			pendingLock.unlock();
		}

		n.shift();

		if (logger.isLoggable(Level.FINE)) {
			logger.fine(eventInspector.toString(node));
		}

		assert n == n.getSubscription().getRoot();

		n.getSubscription().nodeUpdated(eventInspector);
		// TODO unlock
	}

	@SuppressWarnings("unchecked")
	public Root<N> activated(Key key) throws NoSuchSubscription {
		// logger.fine(notifier.toString());

		// MemRootNode node = new HashedRootNode(nodeType, key);
		MemRootNode node = new LinkedRootNode(nodeType, key);

		// activation of a subscription within invokation of Notifier.subscribe?
		if (pendingKey != null && pendingKey.equals(key)) {
			if (pendingSubscription != null) {
				// activating after activation (and probably subsequent
				// deactivation) within a call to Notifier.subscribe ...

				pendingSubscription.activate(node);
			} else {
				// create and register a new subscription
				pendingSubscription = new SubscriptionImpl<N>(key, node, this,
						false);

				SubscriptionImpl<N> s = dataLayer.registerSubscription(key,
						pendingSubscription, pendingSubscriber);

				if (s != pendingSubscription) {
					// TODO: delay it?
					notifier.unsubscribe(key);
				}

				pendingSubscription = s;
			}
		} else {
			// get subscription by key and activate it
			SubscriptionImpl<N> subscription;
			subLock.lock();
			subscription = subscriptions.get(key);
			subLock.unlock();

			if (subscription == null)
				throw new NoSuchSubscription(key);

			assert subscription.getKey() == key;

			subscription.activate(node);
		}

		return node;
	}

	public void deactivated(Key key, Status state) throws NoSuchSubscription {
		logger.fine(notifier.toString());

		// deactivation of a subscription within invokation of
		// Notifier.subscribe?
		if (pendingKey != null && pendingKey.equals(key)) {
			if (pendingSubscription != null) {
				// deactivating after activation within a call to
				// Notifier.subscribe ...

				pendingSubscription.deactivate(state);
			} else {

				pendingSubscription = new SubscriptionImpl<N>(key, null, this);

				SubscriptionImpl<N> s = dataLayer.registerSubscription(key,
						pendingSubscription, pendingSubscriber);

				if (s != pendingSubscription) {
					// TODO: delay it?
					notifier.unsubscribe(key);
				} else {
					pendingSubscription.deactivate(state); // set status
				}

				pendingSubscription = s;
			}

		} else {
			SubscriptionImpl<N> subscription;
			subLock.lock();
			subscription = subscriptions.get(key);
			subLock.unlock();

			if (subscription == null)
				throw new NoSuchSubscription(key);

			assert subscription.getKey() == key;

			subscription.deactivate(state);
		}

	}

	public SubscriptionImpl<N> subscribe(Subscriber subscriber, Key key, Track track) {
		pendingLock.lock();

		if (logger.isLoggable(Level.FINE))
			logger.fine("Subscribe " + subscriber + " to " + notifier
					+ key.toString());

		assert pendingKey == null : "Only one pending subscription is allowed";

		pendingKey = key;
		pendingSubscription = null;
		pendingSubscriber = subscriber;

		boolean hasSubscribed = false;

		try {
			hasSubscribed = notifier.subscribe(key, track);

			logger.fine("hassubscribed = " + hasSubscribed);

			if (hasSubscribed) {
				assert pendingSubscription != null;

				if (pendingSubscription == null) {
					// TODO: remove
					System.out.println("-----------------------------"
							+ Thread.currentThread());
					pendingKey = null;
					return null;
				}

				subLock.lock();
				subscriptions.put(pendingKey, pendingSubscription);
				subLock.unlock();

				assert pendingSubscription.getKey() == pendingKey;

				SubscriptionImpl<N> result = pendingSubscription;
				pendingKey = null;
				return result;
			} else {
				pendingKey = null;
				return null;
			}

		} catch (Exception e) {
			e.printStackTrace();
			hasSubscribed = false;
			pendingKey = null;
			return null;
		} finally {
			pendingLock.unlock();
		}
	}

	public void unsubscribe(Key key) {
		notifier.unsubscribe(key);
	}

	public Root<N> getRoot(Key key) {
		if (key.equals(pendingKey)) {
			// works correcty, however in some cases may
			// lock on a wrong pending key which does
			// not result in an incorrect behaviour
			pendingLock.lock();
			pendingLock.unlock();
			return pendingSubscription.getRoot();
		}

		SubscriptionImpl<N> s = subscriptions.get(key);
		if (s == null)
			return null;

		assert s.getKey() == key;

		return s.getRoot();
	}

	public NodeType getNodeType() {
		return nodeType;
	}

	public double getPriority() {
		return priority;
	}

	public KeyPattern getPattern() {
		return pattern;
	}

	public PatternMatcher getMatcher() {
		return matcher;
	}

	public void cancelWrite(Node node) {
		throw new RuntimeException("Not implemented");
	}

	public DataLayer getDataLayer() {
		return dataLayer;
	}

	public Status getSubscriptionState(Key key) {
		SubscriptionImpl<N> s = subscriptions.get(key);

		if (s == null)
			return null;

		assert s.getKey() == key;

		return s.getStatus();
	}

	@Override
	public String toString() {
		return "NP(" + dataLayer.nodeType + ") " + notifier + " " + pattern
				+ " ";
	}
}
